package com.lzj.jdbc;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class HelloHBaseDQL {
    //获取HBase管理员类
    private Admin admin;
    //获取数据库连接
    private Connection connection;
    //获取表信息
    private Table table;

    /**
     * 根据RK获取表信息
     *
     * @throws IOException
     */
    @Test
    public void getRowData() throws IOException {

        //获取Get
        Get get = new Get(Bytes.toBytes("s205012999"));
        Result rs = table.get(get);
        Cell[] cells = rs.rawCells();
        for (Cell cell : cells) {
            String columnName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
            String columnValue = null;
            if (columnName.equals("money") || columnName.equals("age")) {
                int result = Bytes.toInt(cell.getValueArray());
                columnValue = Integer.toString(result);
            } else {
                columnValue = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
            }
            System.out.println(columnName + ":" + columnValue);
        }
    }

    /**
     * 获取多行数据
     *
     * @throws IOException
     */
    public void getRowDataList() throws IOException {
        List<Get> gets = new ArrayList<>();
        gets.add(new Get(Bytes.toBytes("s001")));
        gets.add(new Get(Bytes.toBytes("s002")));
        gets.add(new Get(Bytes.toBytes("s003")));
        gets.add(new Get(Bytes.toBytes("s004")));
        Result[] rs = table.get(gets);
        for (Result r : rs) {
            Cell[] cells = r.rawCells();
            for (Cell c : cells) {
                String columnName = Bytes.toString(c.getQualifierArray(), c.getQualifierOffset(), c.getQualifierLength());
                String columnValue = null;
                if (columnName.equals("money") || columnName.equals("age")) {
                    int result = Bytes.toInt(c.getValueArray());
                    columnValue = Integer.toString(result);
                } else {
                    columnValue = Bytes.toString(c.getValueArray(), c.getValueOffset(), c.getValueLength());
                }
                System.out.println(columnName + ":" + columnValue);
            }
        }
    }

    /**
     * 扫描整个表空间
     *
     * @throws IOException
     */
    @Test
    public void scanWholeTable() throws IOException {

        Scan scan = new Scan();
        scan.withStartRow(Bytes.toBytes("s01"));
        scan.withStopRow(Bytes.toBytes("s08"));
        ResultScanner scanner = table.getScanner(scan);
        for (Result rs : scanner) {
            System.out.println("rowKey:" + Bytes.toString(rs.getRow()));
            Cell[] cells = rs.rawCells();
            for (Cell cell : cells) {
                String columnName = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
                String columnValue = null;
                if (Arrays.asList("age", "money", "Chinese", "English", "Math").contains(columnName)) {
                    int result = Bytes.toInt(cell.getValueArray());
                    columnValue = Integer.toString(result);
                } else {
                    columnValue = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                }
                System.out.println(columnName + ":" + columnValue);
            }
        }
    }

    @Before
    public void init() throws IOException {
        Configuration configuration = HBaseConfiguration.create();
        this.connection = ConnectionFactory.createConnection(configuration);
        this.admin = connection.getAdmin();
        //获取数据库表信息
        table = this.connection.getTable(TableName.valueOf("lzj:student"));
    }

    @After
    public void destory() throws IOException {
        if (admin != null) {
            admin.close();
        }
    }

}
